/*
 * RMDispatchingLayerProxy.java   0.0.1    Nov 20, 2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.net.ril.rm;

import java.io.Serializable;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.mentalsmash.documentation.Concurrency;
import org.mentalsmash.tazio.commons.TazioRuntimeException;
import org.mentalsmash.tazio.commons.identifiers.NID;
import org.mentalsmash.tazio.commons.identifiers.OID;
import org.mentalsmash.tazio.net.ril.RemoteInvocationMessageHandler;
import org.mentalsmash.tazio.net.ril.RemoteObject;
import org.mentalsmash.tazio.net.ril.RemoteMethod.InvocationSemantics;
import org.mentalsmash.tazio.net.ril.exceptions.FailedMessageDeliveryException;
import org.mentalsmash.tazio.net.ril.exceptions.NoResponseReceivedException;
import org.mentalsmash.tazio.net.ril.exceptions.UnavailableRemoteNodeException;
import org.mentalsmash.tazio.net.ril.impl.AbstractDispatchingLayerProxy;
import org.mentalsmash.tazio.net.ril.messages.MethodInvocationRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.MethodInvocationResponseMessage;
import org.mentalsmash.tazio.net.ril.messages.OnGroupMethodInvocationRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.RILConnectionRequest;
import org.mentalsmash.tazio.net.ril.messages.RILConnectionResponse;
import org.mentalsmash.tazio.net.ril.messages.RILMessage;
import org.mentalsmash.tazio.net.ril.messages.RILRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.RILResponseMessage;
import org.mentalsmash.tazio.net.ril.messages.RegistryContentRequestMessage;
import org.mentalsmash.tazio.net.ril.messages.RegistryContentResponseMessage;
import org.mentalsmash.tazio.net.ril.messages.RemoteObjectRequestMessage;
import org.mentalsmash.tazio.transparency.commons.messages.DisconnectionMessage;
import org.mentalsmash.tazio.transparency.interfaces.InputMessageListener;
import org.mentalsmash.tazio.transparency.interfaces.TazioNetwork;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.internal.Preconditions;

/**
 * @version 0.0.1 Nov 20, 2009
 * @author Andrea Sorbini <as@mentalsmash.org>
 * 
 */
@Concurrency(concurrentAccess = true, specification = "incoming messages handling is "
	+ "serialized until it has been verified that a received message is not an "
	+ "invocation request of a method with OnGroup semantics, in which case locks "
	+ "are released before actual handling of the message. Otherwise they are kept "
	+ "until complete computation of the response message in order to guarantee they're handled "
	+ "in the order dictated by the underlying dispatching layer "
	+ "(i.e. OnGroup invocations are guaranteed to be always handled in the same order "
	+ "they were notified - and hopefully received - by the Dispatching Layer)")
public class RMDispatchingLayerProxy extends AbstractDispatchingLayerProxy
	implements InputMessageListener {

    private static final Logger log = LoggerFactory
	    .getLogger(RMDispatchingLayerProxy.class);

    // private final ServerDispatcher dispatcher;
    private final TazioNetwork network;

    private final Map<NID, OID[]> wrms = new HashMap<NID, OID[]>();
    private final Map<NID, OID[]> disconnectedWRMs = new HashMap<NID, OID[]>();
    
    private final Object wrmsLock = new Object();

    @Inject
    public RMDispatchingLayerProxy(RemoteInvocationMessageHandler handler,
	    TazioNetwork network) {
	super(handler);

	this.network = network;

	this.network.register(RILMessage.class, this);
	this.network.register(DisconnectionMessage.class, this);
    }

    public RMDispatchingLayerProxy(RemoteInvocationMessageHandler handler,
	    TazioNetwork network, long timeout) {
	super(handler, timeout);

	this.network = network;

	this.network.register(RILMessage.class, this);
	this.network.register(DisconnectionMessage.class, this);
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.impl.AbstractDispatchingLayerProxy#
     * actuallySendMessage
     * (org.mentalsmash.tazio.net.ril.messages.RILRequestMessage)
     */
    @Override
    protected void actuallySendMessage(RILRequestMessage msg)
	    throws FailedMessageDeliveryException {

	NID recipientID = this.findRecipient(msg);

	if (recipientID == null) {
	    log.error("Cannot find recipient for message " + msg);
	    throw new FailedMessageDeliveryException(msg);
	}

	try {

	    RILMessage response = (RILMessage) network.send(msg, recipientID);

	    if (!(response instanceof RILResponseMessage)) {
		throw new TazioRuntimeException(
			"Response received wasn't a RILResponseMessage but instead: "
				+ response);
	    } else {
		this.responseReceived((RILResponseMessage) response);
	    }

	} catch (Exception e) {
	    log.debug("Error sending message", e);
	    throw new FailedMessageDeliveryException(msg);
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.impl.AbstractDispatchingLayerProxy#
     * actuallySendMethodInvocationAtMostOnce
     * (org.mentalsmash.tazio.net.ril.messages.MethodInvocationRequestMessage)
     */
    @Override
    protected void actuallySendMethodInvocationAtMostOnce(
	    MethodInvocationRequestMessage msg)
	    throws FailedMessageDeliveryException {

	NID recipientID = this.findRecipient(msg);

	if (recipientID == null) {
	    log.error("Cannot find recipient for message " + msg);
	    throw new FailedMessageDeliveryException(msg);
	}

	try {

	    RILMessage response = (RILMessage) network.send(msg, recipientID);

	    if (!(response instanceof RILResponseMessage)) {
		throw new TazioRuntimeException(
			"Response received wasn't a RILResponseMessage");
	    } else {
		this.responseReceived((RILResponseMessage) response);
	    }

	} catch (Exception e) {
	    log.debug("Error sending message", e);
	    throw new FailedMessageDeliveryException(msg);
	}

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.ril.impl.AbstractDispatchingLayerProxy#
     * actuallySendMethodInvocationToGroup
     * (org.mentalsmash.tazio.net.ril.messages.MethodInvocationRequestMessage)
     */
    @Override
    protected void actuallySendMethodInvocationToGroup(
	    MethodInvocationRequestMessage msg)
	    throws FailedMessageDeliveryException {

	try {

	    this.network.sendToGroup(msg);

	    MethodInvocationResponseMessage fakeResponse = new MethodInvocationResponseMessage(
		    msg.getOID(), null, false);
	    this.responseReceived((RILResponseMessage) fakeResponse);
	} catch (Exception e) {
	    log.debug("Error sending message", e);
	    throw new FailedMessageDeliveryException(msg);
	}

    }

    private NID findRecipient(RILRequestMessage msg) {

	OID objOID = null;

	if (msg instanceof MethodInvocationRequestMessage) {
	    objOID = ((MethodInvocationRequestMessage) msg).getRemoteObject();
	    log.debug("Locating owner of " + objOID + " to invoke "
		    + ((MethodInvocationRequestMessage) msg).getRemoteMethod());
	} else if (msg instanceof RemoteObjectRequestMessage) {
	    objOID = ((RemoteObjectRequestMessage) msg).getRequested();
	    log.debug("Locating owner of " + objOID);
	} else {
	    throw new TazioRuntimeException("unknown type of request to send");
	}

	synchronized (this.wrmsLock) {
	    NID recipient = null;

		log.debug("Looking for owner of "+objOID);
	    for (NID h : this.wrms.keySet()) {
                log.debug("-----------");
                log.debug("WRM "+h+" : ");
                for (OID oid : this.wrms.get(h)){
                    log.debug(oid+"");
                }
                log.debug("-----------");
            }


	    for (NID nid : this.wrms.keySet()) {
		OID[] content = this.wrms.get(nid);
		for (OID oid : content) {
		    if (oid.equals(objOID)) {
			recipient = nid;
			break;
		    }
		}
		/*if (nid != null) {
		    break;
		}*/
	    }

	    log.debug("Owner of " + objOID + " is : " + recipient);

	    return recipient;
	}

	// synchronized (this.wrmsLock) {
	// for (NID nodeId : this.wrms.keySet()) {
	// OID[] content = this.wrms.get(nodeId);
	//
	// for (OID oid : content) {
	// if (oid.equals(objOID)) {
	// return nodeId;
	// }
	// }
	// }
	//
	// Set<NID> deadNodes = new HashSet<NID>();
	// Map<NID, OID[]> updatedContents = new HashMap<NID, OID[]>();
	//
	// for (NID nodeId : this.wrms.keySet()) {
	// try {
	// OID[] cont = this.getUpdatedWRMContent(nodeId);
	// updatedContents.put(nodeId, cont);
	//
	// } catch (UnavailableRemoteNodeException e) {
	// log.error("Cannot retrieve node's registry content : ", e);
	// deadNodes.add(nodeId);
	// }
	//
	// }
	//
	// for (NID nid : updatedContents.keySet()) {
	// this.wrms.put(nid, updatedContents.get(nid));
	// }
	//
	// for (NID nid : deadNodes) {
	// this.wrms.remove(nid);
	// }
	//
	// for (NID nodeId : this.wrms.keySet()) {
	// OID[] content = this.wrms.get(nodeId);
	//
	// for (OID oid : content) {
	// if (oid.equals(objOID)) {
	// return nodeId;
	// }
	// }
	// }
	// }
	//
	// return null;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.dispatching.RMInputMessageListener#messageReceived
     * (org.mentalsmash.tazio.core.messages.TazioMessage)
     */
    @Override
    public <T extends Serializable> Serializable handleMessage(NID sender,
	    T message) {
	
//	System.gc();

	if (message instanceof DisconnectionMessage) {
	    synchronized (wrmsLock) {
		DisconnectionMessage msg = (DisconnectionMessage) message;
		OID[] oids = this.wrms.get(msg.getDisconnectedNode());
		this.wrms.remove(msg.getDisconnectedNode());
		this.disconnectedWRMs.put(msg.getDisconnectedNode(), oids);
		log.debug("Host "+msg.getDisconnectedNode()+" disconnected");
	    }

	    return null;
	}

	// Preconditions.checkArgument(message instanceof RILMessage,
	// "Message must be an instance of TazioMessage");
	RILMessage msg = RILMessage.class.cast(message);

	log.debug("Received message : " + message);

	// System.err.println("Received message : " + message);

	// if (message instanceof MethodInvocationRequestMessage) {
	// MethodInvocationRequestMessage invocMsg =
	// (MethodInvocationRequestMessage) message;
	//
	// boolean onGroup = invocMsg.getRemoteMethod().getSemantics().equals(
	// InvocationSemantics.OnGroup);
	//
	// synchronized (this.wrmsLock) {
	// if (!onGroup && !this.wrms.containsKey(sender)) {
	// try {
	// OID[] content = this.getUpdatedWRMContent(sender);
	// this.wrms.put(sender, content);
	// } catch (UnavailableRemoteNodeException e) {
	// log.error("Cannot retrieve node's registry content : ",
	// e);
	// this.wrms.remove(sender);
	// }
	// ;
	// }
	// }
	//
	// } else {
	// synchronized (this.wrmsLock) {
	// if (!this.wrms.containsKey(sender)) {
	// try {
	// OID[] content = this.getUpdatedWRMContent(sender);
	// this.wrms.put(sender, content);
	// } catch (UnavailableRemoteNodeException e) {
	// log.error("Cannot retrieve node's registry content : ",
	// e);
	// this.wrms.remove(sender);
	// }
	// }
	// }
	//
	// }

	this.lockGroupInvocations();
	RILMessage response = this.handleIncomingMessage(sender,msg);

	return response;
    }

    private RILResponseMessage handleConnectionRequest(RILConnectionRequest req) {
	synchronized (wrmsLock) {
	    NID host = req.getHostNID();
	    OID[] content = req.getRegistryContent();
	    this.wrms.put(host, content);
	    if (this.disconnectedWRMs.containsKey(req.getHostNID())) {
		this.disconnectedWRMs.remove(req.getHostNID());
	    }

	    log.debug("New WRM Connected (" + content.length
		    + " objs) OID: " + host);


	    for (NID h : this.wrms.keySet()) {
	        log.debug("-----------");
 		log.debug("WRM "+h+" : ");
		for (OID o : this.wrms.get(h)){
		    log.debug(o+"");
		}
		log.debug("-----------");
	    }

	    RegistryContentRequestMessage regReq = new RegistryContentRequestMessage();
	    RegistryContentResponseMessage regResp = (RegistryContentResponseMessage) this
		    .handleIncomingMessage(regReq);

	    RILConnectionResponse resp = new RILConnectionResponse(
		    req.getOID(), regResp.getContent());
	    return resp;
	}
    }

    private OID[] getUpdatedWRMContent(NID nodeID)
	    throws UnavailableRemoteNodeException {

	log.error("Updating remote registry content for WRM " + nodeID);

	RegistryContentRequestMessage request = new RegistryContentRequestMessage();
	try {
	    RILMessage resp = (RILMessage) network.send(request, nodeID);
	    if (!(resp instanceof RegistryContentResponseMessage)) {
		throw new TazioRuntimeException(
			"response received wasn't a Registry content response");
	    }

	    OID[] content = ((RegistryContentResponseMessage) resp)
		    .getContent();

	    // synchronized (this.wrmsLock) {
	    // if (this.wrms.containsKey(nodeID)) {
	    // this.wrms.remove(nodeID);
	    // }
	    //
	    // this.wrms.put(nodeID, content);
	    // }

	    return content;
	} catch (Exception e) {
	    log
		    .error(
			    "Exception while retrieving node's registry content: ",
			    e);

	    throw new UnavailableRemoteNodeException(nodeID);
	}

    }

    protected RILMessage handleIncomingMessage(NID sender, RILMessage message) {
	if (message instanceof OnGroupMethodInvocationRequestMessage) {
	    this.unlockGroupInvocations();
	    return this
		    .handleOnGroupInvocationFromClient((OnGroupMethodInvocationRequestMessage) message);
	} else {
	    if (message instanceof RILConnectionRequest) {
		this.unlockGroupInvocations();
		RILResponseMessage resp = this
			.handleConnectionRequest((RILConnectionRequest) message);
		return resp;
	    }

	    if (message instanceof MethodInvocationRequestMessage) {
		MethodInvocationRequestMessage invocReq = (MethodInvocationRequestMessage) message;

		if (invocReq.getRemoteMethod().getSemantics().equals(
			InvocationSemantics.OnGroup)) {

		    RILMessage response = super.handleIncomingMessage(message);
		    this.unlockGroupInvocations();
		    return response;

		} else {
		    this.unlockGroupInvocations();
		    if (!invocReq.getRemoteMethod().isByCopy()) {
			this.registerRemoteArguments(sender,invocReq);
		    }

		    return super.handleIncomingMessage(invocReq);
		}

	    } else if (message instanceof RemoteObjectRequestMessage) {
		this.unlockGroupInvocations();
		return super
			.handleIncomingMessage((RemoteObjectRequestMessage) message);
	    } else if (message instanceof RegistryContentRequestMessage) {
		this.unlockGroupInvocations();
		return super
			.handleIncomingMessage((RegistryContentRequestMessage) message);
	    }
	}

	log.warn("Unknown type of message received : " + message);
	return null;
    }

    private void registerRemoteArguments(NID sender, MethodInvocationRequestMessage invocReq) {
	Serializable[] args = invocReq.getArgs();
	
	OID[] current = this.wrms.get(sender);
	Set<OID> content = new HashSet<OID>();
	for (OID oid : current) {
	    content.add(oid);
	}
	
	for (Serializable arg : args) {
	    if (arg instanceof RemoteObject) {
		OID oid = ((RemoteObject)arg).getOID();
		content.add(oid);
		log.debug("Object "+oid+" on "+sender+" registered");
	    }
	}
	
	this.wrms.put(sender, content.toArray(new OID[0]));
	log.debug("Stored registry's content ("+content.size()+" objs) for "+sender);

	for (NID h : this.wrms.keySet()) {
                log.debug("-----------");
                log.debug("WRM "+h+" : ");
                for (OID o : this.wrms.get(h)){
                    log.debug(o+"");
                }
                log.debug("-----------");
            }
	
    }

    private final Object groupInvocationLock = new Object();
    private int groupInvoking = 0;

    private void lockGroupInvocations() {
	synchronized (groupInvocationLock) {
	    while (groupInvoking > 0) {
		try {
		    groupInvocationLock.wait();
		} catch (InterruptedException e) {

		}
	    }

	    groupInvoking++;
	}
    }

    private void unlockGroupInvocations() {
	synchronized (groupInvocationLock) {
	    groupInvoking--;
	    groupInvocationLock.notifyAll();
	}
    }

    private MethodInvocationResponseMessage handleOnGroupInvocationFromClient(
	    OnGroupMethodInvocationRequestMessage msg) {
	MethodInvocationResponseMessage response = null;

	MethodInvocationRequestMessage request = msg.getRequest();

	try {
	    response = this.sendMethodInvocationToGroup(request);
	} catch (NoResponseReceivedException e) {
	    response = new MethodInvocationResponseMessage(request.getOID(), e,
		    true);
	} catch (FailedMessageDeliveryException e) {
	    response = new MethodInvocationResponseMessage(request.getOID(), e,
		    true);
	}

	return response;
    }

}
